package com.autonavi.collapse.task.db;

import com.autonavi.collapse.task.annotation.TaskComponent;
import com.autonavi.collapse.task.context.utils.NetUtils;
import com.autonavi.collapse.task.leader.LeaderEvent;
import com.autonavi.collapse.task.leader.LeaderEventType;
import com.autonavi.collapse.task.leader.LeaderListener;
import com.autonavi.collapse.task.leader.LeaderSelector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@TaskComponent
public class DbLeaderSelector implements LeaderSelector {

    private static final Logger LOGGER = LoggerFactory.getLogger(DbLeaderSelector.class);
    private static final String SELECT_LEADER = "UPDATE collapse_locker SET time_out=date_add(now(),INTERVAL ? second),owner=? WHERE name=? AND (owner=? OR NOW()>time_out)";
    private static final String FORCE_LEADER = "UPDATE collapse_locker SET time_out=date_add(now(),INTERVAL ? second),owner=?";
    private static final String CHECK_LOCK = "SELECT 1 FROM collapse_locker WHERE name=? FOR UPDATE";
    private static final String CREATE_LOCK = "INSERT INTO collapse_locker(name) VALUES(?)";

    private DataSource dataSource;

    private int timeOut = 30;
    private int heartbeatInterval = 5000;
    private String group;
    private String nodeName;
    private boolean forceLeader = false;

    private volatile boolean leader = false;
    private ScheduledExecutorService scheduledExecutorService;
    private List<LeaderListener> leaderListeners;

    public DbLeaderSelector() {
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r);
            thread.setName("leaderSelector");
            return thread;
        });
        leaderListeners = new ArrayList<>();
    }

    @Override
    public void init(String group, Object dataCenter) {
        this.group = group;
        if (dataCenter instanceof DataSource) {
            dataSource = (DataSource) dataCenter;
        }
    }

    @Override
    public boolean isLeader() {
        return leader;
    }

    @Override
    public void addLeaderListener(LeaderListener leaderListener) {
        leaderListeners.add(leaderListener);
    }

    @Override
    public void start() throws Exception {
        if (nodeName == null) {
            nodeName = NetUtils.getLocalIp();
        }
        checkLock();
        if (forceLeader) {
            forceLeaderNode(nodeName, timeOut);
            leader = true;
        }
        scheduledExecutorService.scheduleAtFixedRate(this::selectLeader, heartbeatInterval, heartbeatInterval, TimeUnit.MILLISECONDS);
    }

    @Override
    public void stop() {
        scheduledExecutorService.shutdownNow();
    }

    private void onLeader(boolean leader) {
        LeaderEvent event;
        if (leader) {
            event = new LeaderEvent(LeaderEventType.TASK);
        } else {
            event = new LeaderEvent(LeaderEventType.LOST);
        }
        for (LeaderListener leaderListener : leaderListeners) {
            try {
                leaderListener.onEvent(event);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
    }

    private void selectLeader() {
        LOGGER.debug("{} send heartbeat", nodeName);
        try {
            int rowAffected = updateLeaderNode(this.group, this.nodeName, timeOut);
            if (rowAffected > 0) {
                if (!leader) {
                    leader = true;
                    onLeader(true);
                    LOGGER.info("{} task leader", nodeName);
                }
            } else {
                if (leader) {
                    leader = false;
                    onLeader(false);
                    LOGGER.info("{} lost leader", nodeName);
                }
            }
        }catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
            leader = false;
            onLeader(false);
            LOGGER.info("{} lost leader", this.nodeName);
        }
    }

    private int updateLeaderNode(String group, String name, int timeOut) throws SQLException {
        try (Connection connection = dataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(SELECT_LEADER)) {
            preparedStatement.setInt(1, timeOut);
            preparedStatement.setString(2, name);
            preparedStatement.setString(3, group);
            preparedStatement.setString(4, name);
            return preparedStatement.executeUpdate();
        }
    }

    private void forceLeaderNode(String name, int timeOut) throws SQLException {
        try (Connection connection = dataSource.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(FORCE_LEADER)){
            preparedStatement.setInt(1, timeOut);
            preparedStatement.setString(2, name);
            preparedStatement.executeUpdate();
        }
    }

    private void checkLock() throws SQLException {
        try (Connection connection = dataSource.getConnection()){
            connection.setAutoCommit(false);
            boolean exist;
            try {
                try (PreparedStatement preparedStatement = connection.prepareStatement(CHECK_LOCK)){
                    preparedStatement.setString(1, group);
                    try (ResultSet resultSet = preparedStatement.executeQuery()){
                        exist = resultSet.next();
                    }
                }
                if (!exist) {
                    try (PreparedStatement preparedStatement = connection.prepareStatement(CREATE_LOCK)){
                        preparedStatement.setString(1, group);
                        preparedStatement.executeUpdate();
                    }
                }
            } finally {
                connection.commit();
            }
        }
    }

    public void setTimeOut(int timeOut) {
        this.timeOut = timeOut;
    }

    public void setHeartbeatInterval(int heartbeatInterval) {
        this.heartbeatInterval = heartbeatInterval;
    }

    public void setForceLeader(boolean forceLeader) {
        this.forceLeader = forceLeader;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public void setNodeName(String nodeName) {
        this.nodeName = nodeName;
    }
}
